package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TradeSkuOrderBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import com.bw.gmall.realtime.common.util.HBaseUtil;
import com.bw.gmall.realtime.common.util.RedisUtil;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.AsyncConnection;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.math.BigDecimal;
import java.time.Duration;
import java.util.Collections;
import java.util.Iterator;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

public class DwsTradeSkuOrderAsyncCacheWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeSkuOrderAsyncCacheWindow().start(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,Constant.DWS_TRADE_SKU_ORDER_WINDOW,4,10093);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {


        //etl
//        dataStreamSource.print();
        SingleOutputStreamOperator<JSONObject> streamOperator = EltStream(dataStreamSource);

        //水位线
        SingleOutputStreamOperator<JSONObject> waterStream = WaterStream(streamOperator);

        // 2. 按照 order_detail_id // 4.防止下游重复计算
        SingleOutputStreamOperator<TradeSkuOrderBean> process = getStreamOperator(waterStream);

        // 5.以sku_id分组
        KeyedStream<TradeSkuOrderBean, String> keyedStream = process.keyBy(x -> x.getSkuId());


        //开窗 聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduce = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean t1, TradeSkuOrderBean t2) throws Exception {

                        t1.setActivityReduceAmount(t1.getActivityReduceAmount().add(t2.getActivityReduceAmount()));
                        t1.setCouponReduceAmount(t1.getCouponReduceAmount().add(t2.getCouponReduceAmount()));
                        t1.setOrderAmount(t1.getOrderAmount().add(t2.getOrderAmount()));
                        t1.setOriginalAmount(t1.getOriginalAmount().add(t2.getOriginalAmount()));

                        return t1;
                    }
                }, new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TradeSkuOrderBean> iterable, Collector<TradeSkuOrderBean> collector) throws Exception {

                        long start = timeWindow.getStart();
                        long end = timeWindow.getEnd();

                        Iterator<TradeSkuOrderBean> iterator = iterable.iterator();
                        TradeSkuOrderBean tradeSkuOrderBean = iterator.next();

                        tradeSkuOrderBean.setStt(DateFormatUtil.tsToDateTime(start));
                        tradeSkuOrderBean.setEdt(DateFormatUtil.tsToDateTime(end));
                        tradeSkuOrderBean.setCurDate(DateFormatUtil.tsToDateTime(System.currentTimeMillis()));
                        collector.collect(tradeSkuOrderBean);
                    }
                });

//        reduce.print();

        // 4. join 维度   异步 + 缓存

        SingleOutputStreamOperator<TradeSkuOrderBean> mapstream = JoinSteam(reduce);


        mapstream.print();

        //写入 doris
//        mapstream.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSinks(Constant.DWS_TRADE_SKU_ORDER_WINDOW));


    }

    private static SingleOutputStreamOperator<TradeSkuOrderBean> JoinSteam(SingleOutputStreamOperator<TradeSkuOrderBean> reduce) {
        return AsyncDataStream.unorderedWait(reduce, new RichAsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {

            private StatefulRedisConnection<String, String> redisAsyncConnection;
            private AsyncConnection hBaseAsyncConnection;

            @Override
            public void open(Configuration parameters) throws Exception {
                redisAsyncConnection = RedisUtil.getRedisAsyncConnection();
                hBaseAsyncConnection = HBaseUtil.getHBaseAsyncConnection();

            }


            @Override
            public void close() throws Exception {
                RedisUtil.closeRedisAsyncConnection(redisAsyncConnection);
                HBaseUtil.closeAsyncConnection(hBaseAsyncConnection);
            }

            @Override
            public void asyncInvoke(TradeSkuOrderBean tradeSkuOrderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {

                //先查询 redis 缓存
                String skuId = tradeSkuOrderBean.getSkuId();

                String skuInfokey = RedisUtil.getRedisKey("dim_sku_info", skuId);

                CompletableFuture.supplyAsync(new Supplier<String>() {

                    @Override
                    public String get() {
                        RedisFuture<String> stringRedisFuture = redisAsyncConnection.async().get(skuInfokey);

                        String dimskuinfo = null;
                        try {

                            dimskuinfo = stringRedisFuture.get();

                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } catch (ExecutionException e) {
                            e.printStackTrace();
                        }

                        return dimskuinfo;
                    }
                }).thenApplyAsync(new Function<String, JSONObject>() {
                    @Override
                    public JSONObject apply(String s) {
                        JSONObject dimSkuInfo = null;

                        //从hbase 取值

                        if (s == null || s.length() == 0) {
                            try {
                                dimSkuInfo = HBaseUtil.getAsyncCells(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId);
                                //存入 redis
                                redisAsyncConnection.async().setex(skuInfokey, 24 * 3600, dimSkuInfo.toJSONString());

                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        } else {
                            //redis 里有数据 ，进行取值

                            dimSkuInfo = JSON.parseObject(s);

                        }

                        return dimSkuInfo;
                    }
                }).thenAccept(new Consumer<JSONObject>() {
                    //进行关联维度
                    @Override
                    public void accept(JSONObject jsonObject) {
                        if (jsonObject != null) {
                            //关联sku  dim_sku_info 获取 sku_name spu_id category3_id

                            tradeSkuOrderBean.setSkuName(jsonObject.getString("sku_name"));
                            tradeSkuOrderBean.setCategory3Id(jsonObject.getString("category3_id"));
                            tradeSkuOrderBean.setSpuId(jsonObject.getString("spu_id"));
                            tradeSkuOrderBean.setTrademarkId(jsonObject.getString("tm_id"));

                        } else {
                            System.out.println("没有查到维度数据:" + "dim_sku_info" + ":" + skuId);

                        }


                         //根据 tm_id dim_base_trademark 获取 tm_name

                        String tm_id = tradeSkuOrderBean.getTrademarkId();

                        String trademarkInfokey = RedisUtil.getRedisKey("dim_base_trademark", tm_id);

                        CompletableFuture.supplyAsync(new Supplier<String>() {

                            @Override
                            public String get() {
                                //获取 redis 数据
                                RedisFuture<String> stringRedisFuture = redisAsyncConnection.async().get(trademarkInfokey);

                                String dimtrademark = null;
                                try {

                                    dimtrademark = stringRedisFuture.get();

                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                } catch (ExecutionException e) {
                                    e.printStackTrace();
                                }

                                return dimtrademark;
                            }
                        }).thenApplyAsync(new Function<String, JSONObject>() {
                            @Override
                            public JSONObject apply(String s) {
                                JSONObject dimtrademark = null;

                                //从hbase 取值

                                if (s == null || s.length() == 0) {
                                    try {
                                        dimtrademark = HBaseUtil.getAsyncCells(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, "dim_base_trademark", tm_id);
                                        //存入 redis
                                        redisAsyncConnection.async().setex(trademarkInfokey, 24 * 3600, dimtrademark.toJSONString());

                                    } catch (IOException e) {
                                        e.printStackTrace();
                                    }
                                } else {
                                    //redis 里有数据 ，进行取值

                                    dimtrademark = JSON.parseObject(s);

                                }

                                return dimtrademark;
                            }
                        }).thenAccept(new Consumer<JSONObject>() {
                            //进行关联维度
                            @Override
                            public void accept(JSONObject jsonObject) {
                                if (jsonObject != null) {
                                   //获取 tm_name
                                    tradeSkuOrderBean.setTrademarkName(jsonObject.getString("tm_name"));


                                } else {
                                    System.out.println("没有查到维度数据:" + "dim_base_trademark" + ":" + tm_id);

                                }




                                //根据 spu_id dim_spu_info 获取 spu_name

                                String spu_id = tradeSkuOrderBean.getSpuId();

                                String spuInfokey = RedisUtil.getRedisKey("dim_spu_info", spu_id);

                                CompletableFuture.supplyAsync(new Supplier<String>() {

                                    @Override
                                    public String get() {
                                        //获取 redis 数据
                                        RedisFuture<String> stringRedisFuture = redisAsyncConnection.async().get(spuInfokey);

                                        String dimspuinfo = null;
                                        try {

                                            dimspuinfo = stringRedisFuture.get();

                                        } catch (InterruptedException e) {
                                            e.printStackTrace();
                                        } catch (ExecutionException e) {
                                            e.printStackTrace();
                                        }

                                        return dimspuinfo;
                                    }
                                }).thenApplyAsync(new Function<String, JSONObject>() {
                                    @Override
                                    public JSONObject apply(String s) {
                                        JSONObject dimspuinfo = null;

                                        //从hbase 取值

                                        if (s == null || s.length() == 0) {
                                            try {
                                                dimspuinfo = HBaseUtil.getAsyncCells(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, "dim_spu_info", spu_id);
                                                //存入 redis
                                                redisAsyncConnection.async().setex(spuInfokey, 24 * 3600, dimspuinfo.toJSONString());

                                            } catch (IOException e) {
                                                e.printStackTrace();
                                            }
                                        } else {
                                            //redis 里有数据 ，进行取值

                                            dimspuinfo = JSON.parseObject(s);

                                        }

                                        return dimspuinfo;
                                    }
                                }).thenAccept(new Consumer<JSONObject>() {
                                    //进行关联维度
                                    @Override
                                    public void accept(JSONObject jsonObject) {
                                        if (jsonObject != null) {
                                            //获取 tm_name
                                            tradeSkuOrderBean.setSpuName(jsonObject.getString("spu_name"));


                                        } else {
                                            System.out.println("没有查到维度数据:" + "dim_spu_info" + ":" + spu_id);

                                        }

                                        //返回结果




                                        //根据 category3_id  dim_base_category3 获取 name,category2_id

                                        String category3_id = tradeSkuOrderBean.getCategory3Id();

                                        String cate3Infokey = RedisUtil.getRedisKey("dim_base_category3", category3_id);

                                        CompletableFuture.supplyAsync(new Supplier<String>() {

                                            @Override
                                            public String get() {
                                                //获取 redis 数据
                                                RedisFuture<String> stringRedisFuture = redisAsyncConnection.async().get(cate3Infokey);

                                                String dimcate3info = null;
                                                try {

                                                    dimcate3info = stringRedisFuture.get();

                                                } catch (InterruptedException e) {
                                                    e.printStackTrace();
                                                } catch (ExecutionException e) {
                                                    e.printStackTrace();
                                                }

                                                return dimcate3info;
                                            }
                                        }).thenApplyAsync(new Function<String, JSONObject>() {
                                            @Override
                                            public JSONObject apply(String s) {
                                                JSONObject dimcate3info = null;

                                                //从hbase 取值

                                                if (s == null || s.length() == 0) {
                                                    try {
                                                        dimcate3info = HBaseUtil.getAsyncCells(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, "dim_base_category3", category3_id);
                                                        //存入 redis
                                                        redisAsyncConnection.async().setex(cate3Infokey, 24 * 3600, dimcate3info.toJSONString());

                                                    } catch (IOException e) {
                                                        e.printStackTrace();
                                                    }
                                                } else {
                                                    //redis 里有数据 ，进行取值

                                                    dimcate3info = JSON.parseObject(s);

                                                }

                                                return dimcate3info;
                                            }
                                        }).thenAccept(new Consumer<JSONObject>() {
                                            //进行关联维度
                                            @Override
                                            public void accept(JSONObject jsonObject) {
                                                if (jsonObject != null) {
                                                    //获取 category2_id
                                                    tradeSkuOrderBean.setCategory3Name(jsonObject.getString("name"));
                                                    tradeSkuOrderBean.setCategory2Id(jsonObject.getString("category2_id"));


                                                } else {
                                                    System.out.println("没有查到维度数据:" + "dim_base_category3" + ":" + category3_id);

                                                }

                                                //返回结果





                                                //根据  category3_id  dim_base_category2 获取 name,category1_id

                                                String category2_id = tradeSkuOrderBean.getCategory2Id();

                                                String cate2Infokey = RedisUtil.getRedisKey("dim_base_category2", category2_id);

                                                CompletableFuture.supplyAsync(new Supplier<String>() {

                                                    @Override
                                                    public String get() {
                                                        //获取 redis 数据
                                                        RedisFuture<String> stringRedisFuture = redisAsyncConnection.async().get(cate2Infokey);

                                                        String dimcate2info = null;
                                                        try {

                                                            dimcate2info = stringRedisFuture.get();

                                                        } catch (InterruptedException e) {
                                                            e.printStackTrace();
                                                        } catch (ExecutionException e) {
                                                            e.printStackTrace();
                                                        }

                                                        return dimcate2info;
                                                    }
                                                }).thenApplyAsync(new Function<String, JSONObject>() {
                                                    @Override
                                                    public JSONObject apply(String s) {
                                                        JSONObject dimcate2info = null;

                                                        //从hbase 取值

                                                        if (s == null || s.length() == 0) {
                                                            try {
                                                                dimcate2info = HBaseUtil.getAsyncCells(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, "dim_base_category2", category2_id);
                                                                //存入 redis
                                                                redisAsyncConnection.async().setex(cate2Infokey, 24 * 3600, dimcate2info.toJSONString());

                                                            } catch (IOException e) {
                                                                e.printStackTrace();
                                                            }
                                                        } else {
                                                            //redis 里有数据 ，进行取值

                                                            dimcate2info = JSON.parseObject(s);

                                                        }

                                                        return dimcate2info;
                                                    }
                                                }).thenAccept(new Consumer<JSONObject>() {
                                                    //进行关联维度
                                                    @Override
                                                    public void accept(JSONObject jsonObject) {
                                                        if (jsonObject != null) {
                                                            //获取 category2_id
                                                            tradeSkuOrderBean.setCategory2Name(jsonObject.getString("name"));
                                                            tradeSkuOrderBean.setCategory1Id(jsonObject.getString("category1_id"));


                                                        } else {
                                                            System.out.println("没有查到维度数据:" + "dim_base_category2" + ":" + category2_id);

                                                        }

                                                        //返回结果





                                                        //根据  category1_id  dim_base_category1 获取 name

                                                        String category1_id = tradeSkuOrderBean.getCategory1Id();

                                                        String cate1Infokey = RedisUtil.getRedisKey("dim_base_category1", category1_id);

                                                        CompletableFuture.supplyAsync(new Supplier<String>() {

                                                            @Override
                                                            public String get() {
                                                                //获取 redis 数据
                                                                RedisFuture<String> stringRedisFuture = redisAsyncConnection.async().get(cate1Infokey);

                                                                String dimcate1info = null;
                                                                try {

                                                                    dimcate1info = stringRedisFuture.get();

                                                                } catch (InterruptedException e) {
                                                                    e.printStackTrace();
                                                                } catch (ExecutionException e) {
                                                                    e.printStackTrace();
                                                                }

                                                                return dimcate1info;
                                                            }
                                                        }).thenApplyAsync(new Function<String, JSONObject>() {
                                                            @Override
                                                            public JSONObject apply(String s) {
                                                                JSONObject dimcate1info = null;

                                                                //从hbase 取值

                                                                if (s == null || s.length() == 0) {
                                                                    try {
                                                                        dimcate1info = HBaseUtil.getAsyncCells(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, "dim_base_category1", category1_id);
                                                                        //存入 redis
                                                                        redisAsyncConnection.async().setex(cate1Infokey, 24 * 3600, dimcate1info.toJSONString());

                                                                    } catch (IOException e) {
                                                                        e.printStackTrace();
                                                                    }
                                                                } else {
                                                                    //redis 里有数据 ，进行取值

                                                                    dimcate1info = JSON.parseObject(s);

                                                                }

                                                                return dimcate1info;
                                                            }
                                                        }).thenAccept(new Consumer<JSONObject>() {
                                                            //进行关联维度
                                                            @Override
                                                            public void accept(JSONObject jsonObject) {
                                                                if (jsonObject != null) {
                                                                    //获取 category1_id
                                                                    tradeSkuOrderBean.setCategory1Name(jsonObject.getString("name"));


                                                                } else {
                                                                    System.out.println("没有查到维度数据:" + "dim_base_category1" + ":" + category1_id);

                                                                }

                                                                //返回结果
                                                                resultFuture.complete(Collections.singletonList(tradeSkuOrderBean));



                                                            }
                                                        });


                                                    }
                                                });



                                            }
                                        });

                                    }
                                });


                            }
                        });



                    }
                });


            }
        }, 1000, TimeUnit.SECONDS);
    }

    private static SingleOutputStreamOperator<TradeSkuOrderBean> getStreamOperator(SingleOutputStreamOperator<JSONObject> waterStream) {
         return waterStream.keyBy(x -> x.getString("id")).process(new ProcessFunction<JSONObject, TradeSkuOrderBean>() {

            //处理 回撤流
            MapState<String, BigDecimal> state;


            @Override
            public void open(Configuration parameters) {
                MapStateDescriptor<String, BigDecimal> stateDescriptor = new MapStateDescriptor<>("state", String.class, BigDecimal.class);
                state = getRuntimeContext().getMapState(stateDescriptor);


            }

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, TradeSkuOrderBean>.Context context, Collector<TradeSkuOrderBean> collector) throws Exception {

                //逻辑 ： 来到的数据 先存入 状态中，然后下一次来的数据 减去 上次的值 在存入 状态， 状态里相加就是 没有回撤流的结果

                //取状态的值 若 状态为空 设为 0   模拟从第一次 取状态的值
                BigDecimal originalAmount = state.get("split_original_amount");
                BigDecimal activityReduceAmount = state.get("splitActivityAmount");
                BigDecimal couponReduceAmount = state.get("splitCouponAmount");
                BigDecimal totalorderAmount = state.get("splitTotalAmount");

                //避免为空
                totalorderAmount = totalorderAmount == null ? new BigDecimal(0) : totalorderAmount;
                originalAmount = originalAmount == null ? new BigDecimal(0) : originalAmount;
                activityReduceAmount = activityReduceAmount == null ? new BigDecimal(0) : activityReduceAmount;
                couponReduceAmount = couponReduceAmount == null ? new BigDecimal(0) : couponReduceAmount;


                //获取 数据
                Long ts = jsonObject.getLong("ts");

                String skuId = jsonObject.getString("sku_id");
                String id = jsonObject.getString("id");

                String skuName = jsonObject.getString("sku_name");
                BigDecimal splitTotalAmount = jsonObject.getBigDecimal("split_total_amount");
                BigDecimal splitActivityAmount = jsonObject.getBigDecimal("split_activity_amount");
                BigDecimal splitCouponAmount = jsonObject.getBigDecimal("split_coupon_amount");

                BigDecimal sku_num = jsonObject.getBigDecimal("sku_num");
                BigDecimal orderPrice = jsonObject.getBigDecimal("order_price");
                //乘法
                BigDecimal split_original_amount = orderPrice.multiply(sku_num);



                // 获取 状态内的数据 进行 去除 回撤流 并 返回流
                collector.collect(TradeSkuOrderBean.builder()
                        .skuId(skuId)
                        .orderAmount(splitTotalAmount.subtract(totalorderAmount))
                        .activityReduceAmount(splitActivityAmount.subtract(activityReduceAmount))
                        .couponReduceAmount(splitCouponAmount.subtract(couponReduceAmount))
                        .skuName(skuName)
                        .originalAmount(split_original_amount.subtract(originalAmount))
                        .curDate(DateFormatUtil.tsToDateTime(ts))
                                .ts(ts)

                        .orderDetailId(id)
                        .build());

                //将本次数据 提交到 状态中
                state.put("split_original_amount", split_original_amount);
                state.put("splitActivityAmount", splitActivityAmount);
                state.put("splitCouponAmount", splitCouponAmount);
                state.put("splitTotalAmount", splitTotalAmount);
            }
        });

    }

    private static  SingleOutputStreamOperator<JSONObject> WaterStream(SingleOutputStreamOperator<JSONObject> streamOperator) {
        return streamOperator.assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getLong("ts")).withIdleness(Duration.ofSeconds(1)));
    }

    private static SingleOutputStreamOperator<JSONObject> EltStream(DataStreamSource<String> dataStreamSource) {
         return dataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {

                if (s != null) {

                    JSONObject jsonObject = JSON.parseObject(s);

                    if (jsonObject != null) {

                        String skuId = jsonObject.getString("sku_id");
                        Long ts = jsonObject.getLong("ts");
                        if (skuId != null && ts != null) {

                            jsonObject.put("ts", ts * 1000);

                            collector.collect(jsonObject);
                        }
                    }
                }
            }
        });
    }
}
